package com.quanyan.mq.mq.producer;

import com.google.common.base.Strings;

import com.alibaba.fastjson.JSON;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendCallback;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.client.producer.SendStatus;
import com.alibaba.rocketmq.common.message.Message;
import com.quanyan.mq.client.reqeust.MqRequestMsg;
import com.quanyan.mq.client.reqeust.SendCallbackListener;
import com.quanyan.mq.client.response.MqResponse;
import com.quanyan.mq.enums.MqSendStatus;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @Author Hugo.Wwg
 * @Since 2016-06-07
 */
public abstract class AbstractMqProducer {

    private static final Logger log = LoggerFactory.getLogger(AbstractMqProducer.class);

    private DefaultMQProducer producer;

    private String producerGroup;

    private String nameServerAddress;

    private String instanceName;


    void initProducer() throws MQClientException {
        producer = new DefaultMQProducer(getProducerGroup());//保证应用内唯一
        producer.setNamesrvAddr(getNameServerAddress());
        producer.setInstanceName(getInstanceName());//保证应用内唯一
    }

    public void start() {
        try {
            initProducer();
            producer.start();

        } catch (MQClientException e) {
            log.error(e.getMessage());
        }

        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
            public void run() {
                producer.shutdown();
                log.info("producer调用addShutdownHook方法优雅关闭 ！！！");
            }
        }));
    }


    public MqResponse syncSend(MqRequestMsg mqRequestMsg) {

        Message msg = buildMsg(mqRequestMsg);
        msg.getBody();

        SendResult sendResult = null;
        MqResponse mqResponse = new MqResponse();
        mqResponse.setMqSendStatus(MqSendStatus.FAIL);

        try {
            sendResult = producer.send(msg);//send方法支持内部重试
            if (sendResult == null) {
                return mqResponse;
            }
            SendStatus sendStatus = sendResult.getSendStatus();

            if (sendStatus.equals(SendStatus.SEND_OK)) {
                mqResponse.setMqSendStatus(MqSendStatus.SEND_OK);
            }

            log.info("send message to " + msg.getTopic() + " key: " + mqRequestMsg.getMsgId() +
                    "sendResult : " + com.alibaba.fastjson.JSON.toJSONString(sendResult));

        } catch (Exception e) {

            //如果消息不可丢,必须重发,或者存储到Cache/Db中,由定时器调用重发.
            log.error("send message to " + msg.getTopic() + " key: " + mqRequestMsg.getMsgId() +
                    "sendResult : " + com.alibaba.fastjson.JSON.toJSONString(sendResult) +
                    "sendException : " + e.getMessage());
        }

        return mqResponse;
    }

    public void asyncSend(MqRequestMsg mqRequestMsg,
                          final SendCallbackListener sendCallbackListener) {

        Message msg = buildMsg(mqRequestMsg);

        try {
            producer.send(msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.info(JSON.toJSONString(sendResult.getSendStatus()));
                    if (null == sendCallbackListener) {
                        return;
                    }
                    sendCallbackListener.onSuccess(JSON.toJSONString(sendResult.getSendStatus()));
                }

                @Override
                public void onException(Throwable e) {
                    log.error(e.getMessage());
                    if (null == sendCallbackListener) {
                        return;
                    }
                    sendCallbackListener.onException(e);
                }
            });
            log.info("send message to " + msg.getTopic() + " key: " + mqRequestMsg.getMsgId());

        } catch (Exception e) {
            sendCallbackListener.onException(e);
            e.printStackTrace();
            log.error("send message to " + msg.getTopic() + " key: " + mqRequestMsg.getMsgId() +
                    "sendException : " + e.getMessage());
        }
    }


    public Message buildMsg(MqRequestMsg mqRequestMsg) {
        Message msg = new Message();
        msg.setTopic(mqRequestMsg.getTopic());
        msg.setBody(mqRequestMsg.getBody());
        msg.setFlag(mqRequestMsg.getFlag());
        msg.setKeys(mqRequestMsg.getMsgId()); //每个消息在业务的唯一标识码

        if (!Strings.isNullOrEmpty(mqRequestMsg.getTag())) {
            msg.setTags(mqRequestMsg.getTag());
        }
        return msg;
    }


    public void shutdown() {
        producer.shutdown();
    }

    protected abstract String getProducerGroup();

    protected abstract String getNameServerAddress();

    protected abstract String getInstanceName();

}
